#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2024/10/31 17:27
# @Author  : 王凯
# @File    : main_run.py
# @Project : scrapy_spider
import concurrent.futures
import datetime
import time

from loguru import logger

from apps.pdd.pdd.clean.clean_to_result import CleanToResult
from apps.pdd.pdd.clean.clean_to_task import CleanToTask
from utils.db.redisdb import RedisDB


class PDDCleanRun(CleanToTask):
    def __init__(self, version=None):
        super().__init__()
        self.result_cls = CleanToResult(version)
        self.redis_ab = RedisDB()

    @logger.catch
    def run_by_ids(self, mall_ids):
        mall_ids = [i for i in mall_ids if i]
        if mall_ids:
            super().run_by_ids(mall_ids)
            mall_ids_str = ",".join([f"'{i}'" for i in mall_ids])
            is_all_mall_ids = self.to_db.find(
                f"select mall_id from net_pdd_shop_info where mall_id in ({mall_ids_str}) and (is_all = 1 or flag = -1)",
                to_json=True,
            )
            if is_all_mall_ids:
                self.result_cls.run_by_ids([i["mall_id"] for i in is_all_mall_ids])

    def run(self, time_range=None):
        if time_range is None:
            time_range = datetime.datetime.now().strftime("%Y-%m-%d")
        mall_ids = self.to_db.find(
            f"select distinct mall_id from net_pdd_proxy_log where create_time > '{time_range}'",
            to_json=True,
        )
        mall_ids = [i["mall_id"] for i in mall_ids if i["mall_id"]]
        batch_size = 100
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            for i in range(0, len(mall_ids), batch_size):
                sub_mall_ids = mall_ids[i:i + batch_size]
                executor.submit(self.run_by_ids, sub_mall_ids)

    def put_task_to_redis(self, time_range):
        if time_range is None:
            time_range = datetime.datetime.now().strftime("%Y-%m-%d")
        mall_ids = self.to_db.find(
            f"select distinct mall_id from net_pdd_proxy_log where create_time > '{time_range}'",
            to_json=True,
        )
        mall_ids = [i["mall_id"] for i in mall_ids if i["mall_id"]]
        self.redis_ab.sadd("pdd_clean_task", mall_ids)

    def run_by_redis_batch(self):
        while True:
            mall_ids = self.redis_ab.sget("pdd_clean_task", 100)
            if mall_ids:
                self.run_by_ids(mall_ids)
            else:
                break

    def run_by_redis(self, time_range=None):
        self.put_task_to_redis(time_range)
        max_workers = 10
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            for i in range(max_workers):
                executor.submit(self.run_by_redis_batch)

    # 释放数据
    def release_data(self):
        sql = """
        update net_pdd_shop_info
        set out_state  = 0,
            opera_time = null,
            owner      = null
        where (out_state = 1 or out_state = 3)
          and opera_time < subdate(now(), INTERVAL 24 HOUR)
          and owner is not null
        """
        pass


if __name__ == "__main__":
    test = PDDCleanRun()
    test.run_by_redis(time_range='2025-02-06 21:20')
